package com.aididi.consumer;

import com.aididi.constant.Constants;
import com.aididi.entity.Order;
import com.aididi.util.JsonUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Argument;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * @Auther: zx.xu
 * @Description:    死信队列----》消息接收
 */
@Component
@Slf4j
public class DeadConsumer {

    /**
     *	@Payload(消息体)	    @Headers(消息头)
     */
    @RabbitListener(bindings = {@QueueBinding(//死信队列
            value = @Queue(value = "order_dead_queue",//队列名称
                    durable = "true"),//是否持久化
            //死信路由键
            exchange = @Exchange(value = "order_dead_exchange", //交换机名称
                    durable = "true",//是否持久化
                    type = ExchangeTypes.FANOUT, //消息路由规则
                    ignoreDeclarationExceptions = Constants.IGNOREDECEXCEPTION), // 忽略声明异常
            key = "order_dead_routing_key.#" // 绑定的路由键或模式。
    )}
    )
    @RabbitHandler
    public void  receive(@Payload String msg,
                         @Headers Map<String,Object> map,
                         Channel channel){
        Order order = JsonUtil.covertJsonToObject(msg, Order.class);
        log.info("-----我是死信接受者-----");
        log.info("订单id"+ order.getId());
        //消息唯一标识 ID   DELIVERY_TAG， 是一个单调递增的正整数，delivery tag 的范围仅限于 Channel
        Long tag = (Long) map.get(AmqpHeaders.DELIVERY_TAG);

        try {
            /**
             * 消费端如何限流
             * 当海量消息瞬间推送过来，单个客户端无法同时处理那么多数据，严重会导致系统宕机。这时，需要削峰。
             * RabbitMQ提供了一种qos(服务质量保证)功能。即在非自动确认消息的前提下(非ACK)，
             * 如果一定数目的消息(通过基于consume或者channel设置qos的值)未被确认前，不进行消费新的消息
             *basicQos方法参数：
             * prefetchSize：0，单条消息大小限制，0代表不限制
             * prefetchCount：一次性消费的消息数量。会告诉 RabbitMQ 不要同时给一个消费者推送多于 N 个消息，即一旦有 N 个消息还没有 ack，则该 consumer 将 block 掉，直到有消息 ack。
             * global：true、false 是否将上面设置应用于 channel，简单点说，就是上面限制是 channel 级别的还是 consumer 级别。当我们设置为 false 的时候生效，设置为 true 的时候没有了限流功能，因为 channel 级别尚未实现。
             * 注意：prefetchSize 和 global 这两项，rabbitmq 没有实现，暂且不研究。特别注意一点，prefetchCount 在 no_ask=false 的情况下才生效，即在自动应答的情况下这两个值是不生效的。
             *
             */
            //channel.basicQos(0,1,false);
            /**
             * 消费端消息幂等性保障
             *  1.存redis
             *  2.通过消息唯一id查询
             */
            //1.先查询redis是否存在消息，存在则该消息已被消费过,直接return
            //单消费者例子从string取
            //以消息唯一 id为 key，订单状态为value，存入redis
            /*String messageRedisValue = redisUtil.get(order.id());
            if (messageRedisValue.equals(order.id())) {
                return;
            }*/
            //多消费者例子从list取
            /*List<String> messageRedisValue = redisUtil.lrange("queueName4");
            if (messageRedisValue.contains(order.getId())) {
                return;
            }*/

            //2.由于生产者在发送时生成的唯一id，可通过消息唯一id查询队列里是否存在这条消息，如果不存在，证明消息已被消费过了，队列此时已移除了该消息，直接return
            //可能存在高并发问题，这里不太推荐
            /* if(StringUtils.isEmpty(order.getId())){
                return;
            }*/

            //将id存入string中(单消费者场景)：
            //redisUtil.set("queueName4",order.getId());//以队列为key，id为value
            //将id存入list中(多消费者场景)：
            //redisUtil.lpush("queueName4",order.getId());//存入list
            /**
             *  手工ack认证
             *  第一个参数是标识切勿id
             *  第二个是  是否批量.true,  为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
             */
            channel.basicAck(tag,false);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

}
